package com.suhe.stock.service;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.util.NumberUtil;
import com.suhe.stock.entity.StockKLine;
import jakarta.persistence.EntityManager;
import jakarta.persistence.PersistenceContext;
import jakarta.persistence.Query;
import jakarta.transaction.Transactional;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;

import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

@Service
@RequiredArgsConstructor
public class StockKLineService {

    @PersistenceContext
    private EntityManager entityManager;

    @Transactional
    public void batchReplaceInto(List<StockKLine> list) {

        if (CollectionUtil.isEmpty(list)) return ;

        ArrayBlockingQueue<Double> ma5 = new ArrayBlockingQueue<>(5); // 前5天的收盘价
        ArrayBlockingQueue<Double> ma10 = new ArrayBlockingQueue<>(10); // 前10天的收盘价
        ArrayBlockingQueue<Double> ma20 = new ArrayBlockingQueue<>(20); // 前20天的收盘价
        ArrayBlockingQueue<Double> ma30 = new ArrayBlockingQueue<>(30); // 前30天的收盘价
        ArrayBlockingQueue<Double> ma60 = new ArrayBlockingQueue<>(60); // 前30天的收盘价
        ArrayBlockingQueue<Double> ma120 = new ArrayBlockingQueue<>(120); // 前30天的收盘价

        list.stream().sorted(Comparator.comparing(StockKLine::getDate)) // 按日期排序
                .forEach(item -> {

                    item.setMa5(ma5.isEmpty() ? item.getClose() : NumberUtil.roundHalfEven(Double.valueOf(ma5.stream().mapToDouble(Double::doubleValue).average().orElse(0)), 2).doubleValue());
                    item.setMa10(ma10.isEmpty() ? item.getClose() : NumberUtil.roundHalfEven(Double.valueOf(ma10.stream().mapToDouble(Double::doubleValue).average().orElse(0)), 2).doubleValue());
                    item.setMa20(ma20.isEmpty() ? item.getClose() : NumberUtil.roundHalfEven(Double.valueOf(ma20.stream().mapToDouble(Double::doubleValue).average().orElse(0)), 2).doubleValue());
                    item.setMa30(ma30.isEmpty() ? item.getClose() : NumberUtil.roundHalfEven(Double.valueOf(ma30.stream().mapToDouble(Double::doubleValue).average().orElse(0)), 2).doubleValue());
                    item.setMa60(ma60.isEmpty() ? item.getClose() : NumberUtil.roundHalfEven(Double.valueOf(ma60.stream().mapToDouble(Double::doubleValue).average().orElse(0)), 2).doubleValue());
                    item.setMa120(ma120.isEmpty() ? item.getClose() : NumberUtil.roundHalfEven(Double.valueOf(ma120.stream().mapToDouble(Double::doubleValue).average().orElse(0)), 2).doubleValue());

                    if (!ma5.offer(item.getClose())) {
                        try {
                            ma5.take();
                            ma5.offer(item.getClose());
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                    if (!ma10.offer(item.getClose())) {
                        try {
                            ma10.take();
                            ma10.offer(item.getClose());
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                    if (!ma20.offer(item.getClose())) {
                        try {
                            ma20.take();
                            ma20.offer(item.getClose());
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                    if (!ma30.offer(item.getClose())) {
                        try {
                            ma30.take();
                            ma30.offer(item.getClose());
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                    if (!ma60.offer(item.getClose())) {
                        try {
                            ma60.take();
                            ma60.offer(item.getClose());
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                    if (!ma120.offer(item.getClose())) {
                        try {
                            ma120.take();
                            ma120.offer(item.getClose());
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                });

        // 清理资源
        ma5.clear();
        ma10.clear();
        ma20.clear();
        ma30.clear();
        ma60.clear();
        ma120.clear();

        ListUtil.partition(list, 500)
                .stream().forEach(items -> {
            this.execBatchReplaceInto(items);
        });
    }

    @Transactional(Transactional.TxType.REQUIRES_NEW)
    public void execBatchReplaceInto(List<StockKLine> list) {

        if (CollectionUtil.isEmpty(list)) return;

        // 拼接sql
        StringBuilder sqlBuffer = new StringBuilder("REPLACE INTO stock_k_line (code, name, date, open, close, high, low, volume, turnover, amplitude, changes, up, up_rate, ma5, ma10, ma20, ma30, ma60, ma120) VALUES ");

        AtomicInteger phdIdx = new AtomicInteger(1);
        list.stream().forEach(item -> {
            sqlBuffer.append(" (?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append(", ?").append(phdIdx.getAndIncrement())
                    .append("),");
        });

        String sql = sqlBuffer.toString();
        sql = sql.substring(0, sql.length() - 1);

        Query query = entityManager.createNativeQuery(sql);

        // 设置参数
        AtomicInteger paramIdx = new AtomicInteger(1);
        list.stream().forEach(item -> {
            query.setParameter(paramIdx.getAndIncrement(), item.getCode())
                    .setParameter(paramIdx.getAndIncrement(), item.getName())
                    .setParameter(paramIdx.getAndIncrement(), item.getDate())
                    .setParameter(paramIdx.getAndIncrement(), item.getOpen())
                    .setParameter(paramIdx.getAndIncrement(), item.getClose())
                    .setParameter(paramIdx.getAndIncrement(), item.getHigh())
                    .setParameter(paramIdx.getAndIncrement(), item.getLow())
                    .setParameter(paramIdx.getAndIncrement(), item.getVolume())
                    .setParameter(paramIdx.getAndIncrement(), item.getTurnover())
                    .setParameter(paramIdx.getAndIncrement(), item.getAmplitude())
                    .setParameter(paramIdx.getAndIncrement(), item.getChanges())
                    .setParameter(paramIdx.getAndIncrement(), item.getUp())
                    .setParameter(paramIdx.getAndIncrement(), item.getUpRate())
                    .setParameter(paramIdx.getAndIncrement(), item.getMa5())
                    .setParameter(paramIdx.getAndIncrement(), item.getMa10())
                    .setParameter(paramIdx.getAndIncrement(), item.getMa20())
                    .setParameter(paramIdx.getAndIncrement(), item.getMa30())
                    .setParameter(paramIdx.getAndIncrement(), item.getMa60())
                    .setParameter(paramIdx.getAndIncrement(), item.getMa120());
        });

        // 执行sql
        query.executeUpdate();

    }

    public List<StockKLine> findByCodeOrderByDate(String code) {
        return entityManager.createQuery("select s from StockKLine s where s.code = ?1 order by s.date", StockKLine.class)
                .setParameter(1, code)
                .getResultList();
    }

}
